package com.zsj.hello.six;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.MessageProperties;
import com.zsj.hello.util.RabbimqFactoryUtil;
import org.springframework.scheduling.annotation.Async;

import java.io.IOException;
import java.util.Scanner;
import java.util.UUID;
import java.util.concurrent.TimeoutException;

/**
 * @author https://gitee.com/zhengshengjun
 * @date 2023/8/31.
 * <p>
 * 发布确认
 * 1.单个确认(同步)
 * 2.批量确认(同步)
 * 3.异步批量确认
 */
public class Task04 {
//    public static final String QUEUE_NAME = "simple";

    public static final int MESSAGE_COUNT = 1001;
    public static String QUEUE_NAME= "test_queue";

    public static void main(String[] args) throws Exception {

        //单个确认 1814ms
        //publicMessageIndividually();

        //批量确认 96ms
        //publicMessageByBatch();

        //异步批量确认 49ms
        publicMessageAsync();

    }


    //单个确认
    public static void publicMessageIndividually() throws Exception {
        Channel channel = RabbimqFactoryUtil.getRabbitMQChannel();
        //开启发布确认状态
        channel.confirmSelect();
        //随机生成队列名称
        String QUEUE_NAME = UUID.randomUUID().toString();
        //队列声明并且持久化
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);

        //设置时间
        long begin = System.currentTimeMillis();


//        Scanner scanner = new Scanner(System.in);
        //发布消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            //设置消息
            String message = (i + 1) + "条消息";
            //消息持久化
            channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
            //等候确认
            if (channel.waitForConfirms()) {
                System.out.println("消息发送成功! 消息被保存到磁盘中");
            }
        }

        long endTime = System.currentTimeMillis();
        System.out.println("此次发送" + MESSAGE_COUNT + "条消息总共用时" + (endTime - begin) + "毫秒");

    }

    //批量确认
    public static void publicMessageByBatch() throws Exception {
        Channel channel = RabbimqFactoryUtil.getRabbitMQChannel();
        String QUEUE_NAME = UUID.randomUUID().toString();
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        channel.confirmSelect();

        long begin = System.currentTimeMillis();

        //设置批量确认间隔
        int batchSize = 100;


        for (int i = 1; i < MESSAGE_COUNT; i++) {
            String message = i + "条消息";
            channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());

            //每发布100条确认一下
            if (i % batchSize == 0) {
                if (channel.waitForConfirms()) {
                    System.out.println("100条消息发布确认成功");
                }
            }

        }

        long end = System.currentTimeMillis();

        System.out.println("发送" + (MESSAGE_COUNT - 1) + "条批量消息用时" + (end - begin) + "毫秒");
    }

    //异步确认
    public static void publicMessageAsync() throws Exception {
        Channel channel = RabbimqFactoryUtil.getRabbitMQChannel();
//        QUEUE_NAME = UUID.randomUUID().toString();
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        channel.confirmSelect();
        long begin = System.currentTimeMillis();

        /*
        1.消息的标识
        2.是否为批量确认
         */
        ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
            System.out.println("确认的消息" + deliveryTag);
        };

        ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
            System.out.println("未确认的消息" + deliveryTag);
        };

        //声明监听器(异步)
        channel.addConfirmListener(ackCallback, nackCallback);

        //批量发送消息
        for (int i = 1; i < MESSAGE_COUNT; i++) {
            String message = i + " ";
            channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
        }
 

        long end = System.currentTimeMillis();
        System.out.println("发送" + (MESSAGE_COUNT - 1) + "条异步发布确认消息用时" + (end - begin) + "毫秒");
    }

}
